#include "FDBusServlet.h"

#include <common_base/fdbus.h>
#include <common_base/CFdbProtoMsgBuilder.h>
#include <common_base/cJSON/cJSON.h>
#include <common_base/CFdbCJsonMsgBuilder.h>

#include "ILogger.h"
void printMetadata(FdbObjectId_t obj_id, const CFdbMsgMetadata *metadata)
{
    uint64_t time_c2s;
    uint64_t time_s2r;
    uint64_t time_r2c;
    uint64_t time_total;
    CFdbMessage::parseTimestamp(metadata, time_c2s, time_s2r, time_r2c, time_total);
 
    LogI("OBJ %d , client->server: %lu, arrive->reply: %lu, reply->receive: %lu, total: %lu\n",
                obj_id, time_c2s, time_s2r, time_r2c, time_total);
}

enum EMessageId
{
    REQ_METADATA,
    REQ_RAWDATA,  
};

class RawMessage : public IFdbParcelable{
public:
    void serialize(CFdbSimpleSerializer &serializer) const
    {
        serializer << size << mPrivateInfo;
    }
    void deserialize(CFdbSimpleDeserializer &deserializer)
    {
        deserializer >> size >> mPrivateInfo;
    }
    void put(const void *data,int size) {
        int count = size / 64;
        int left = size % 64;
        int offset = 0;
        for (int i = 0; i < count; i++){
            CFdbByteArray<64> *arr = mPrivateInfo.Add();
            memcpy((char*)arr->buffer(),(char*)data+offset,64);
            offset += 64;
        }
        if (left) {
            CFdbByteArray<64> *arr = mPrivateInfo.Add();
            memcpy((char*)arr->buffer(),(char*)data+offset,left);
            offset += left;
        }
        this->size = size;
    }
    int size;
    void get(void *data,int size) {
        int count = size / 64;
        int left = size % 64;
        int offset = 0;
        if ((count + (left ? 1 : 0) )== mPrivateInfo.size())
        {
            int i = 0;
            for (; i < count;i++){
                 memcpy((char*)data+offset,mPrivateInfo.pool().at(i).buffer(),64);
                offset += 64;
            }
            if (left)
             memcpy((char*)data+offset,mPrivateInfo.pool().at(i).buffer(),left);
        }
           
    }
    CFdbParcelableArray<CFdbByteArray<64>> mPrivateInfo;
protected:
    void toString(std::ostringstream &stream) const
    {
        stream << "size:" << size
               << ", mPrivateInfo: ";
        mPrivateInfo.format(stream);
    }
};

class FDBusSession : public ISession {
public:
    CBaseJob::Ptr &msg_ref;
    FDBusSession( CBaseJob::Ptr &msg_ref):msg_ref(msg_ref){

    }
    virtual void reply(const void *data,int len) {
        CFdbParcelableArray<RawMessage> raws;
        auto msg = castToMessage<CBaseMessage *>(msg_ref);
        RawMessage *meesage = raws.Add();
        meesage->put(data,len);
        CFdbParcelableBuilder builder(raws);
        msg->reply(msg_ref, builder);
    }
    virtual void close() {
         
    }
};

class FDBusSessionClient : public ISession {
public:
    CBaseClient &client;
    FDBusSessionClient(CBaseClient &client):client(client){

    }
     virtual void reply(const void *data,int len) {
        CFdbParcelableArray<RawMessage> raws;
        RawMessage *meesage = raws.Add();
        meesage->put(data,len);
        CFdbParcelableBuilder builder(raws);
        client.invoke(REQ_RAWDATA, builder);
    }
    virtual void close() {
         
    }
};
class FDBusClient : public CBaseClient
{
public:
    FDBusClient(const char *name,FDBusClientServlet*client, CBaseWorker *worker = 0)
        : client(client),CBaseClient(name, worker)
    {
    }

    void onOnline(const CFdbOnlineInfo& info){
        if (client) {
            client->onState(OPEN,0);
        }
        LogI("onOnline\n");
    }
    void onOffline(const CFdbOnlineInfo& info)
    {
        if (client) {
            client->onState(CLOSE,0);
        }
    }
    void onBroadcast(CBaseJob::Ptr &msg_ref)
    {
        handleMessage(msg_ref);
    }
    void onGetEvent(CBaseJob::Ptr &msg_ref){
        handleMessage(msg_ref);
    }
    void onReply(CBaseJob::Ptr &msg_ref){
        handleMessage(msg_ref);
    }
    void onStatus(CBaseJob::Ptr &msg_ref, int32_t error_code, const char *description)
    {
        if (client) {
            client->onState(ERROR,error_code);
        }
    }
private:
    FDBusClientServlet *client;

    void handleMessage(CBaseJob::Ptr &msg_ref) {
        auto msg = castToMessage<CBaseMessage *>(msg_ref);
        LogI("response is receieved. sn: %d\n", msg->sn());
        /* print performance statistics */
        printMetadata(objId(), msg->metadata());

        switch (msg->code())
        {
            case REQ_METADATA:
            {
               
            }
            break;
            case REQ_RAWDATA:
            {
                if (msg->isStatus())
                {
                    int32_t error_code;
                    std::string reason;
                    if (!msg->decodeStatus(error_code, reason))
                    {
                        FDB_LOG_E("onReply: fail to decode status!\n");
                        return;
                    }
                    LogI("CLIENT::onReply(): status is received: msg code: %d, error_code: %d, reason: %s\n",
                          msg->code(), error_code, reason.c_str());
                    return;
                }
                
                CFdbParcelableArray<RawMessage> raw;
                CFdbParcelableParser parser(raw);
                if (!msg->deserialize(parser))
                {
                    LogI("onReply: fail to decode from simple parser!\n");
                    return;
                }
          
                std::ostringstream stream;
                (void)raw.format(stream);
                LogI("CLIENT::onReply: %s\n", stream.str().c_str());
 
                for (auto pit = raw.pool().begin(); pit != raw.pool().end(); ++pit)
                {
                    std::string content;
                    content.resize(pit->size);
                   ((RawMessage*)&(*pit))->get(&content[0],pit->size);
                    FDBusSessionClient session(*this);
                    if (client) {
                        client->onMessage(&session,&content[0],content.size());
                    }
                }
 
            }
            break;
            default:
            break;
        }
    }
};

void FDBusClientServlet::on(SERVLET_STATUS type,std::function<int(ISession *session,void*,int)> handler)
{
    msgidHandlers[type] = handler;
}
void FDBusClientServlet::request(const void *data,int len)
{
    CFdbParcelableArray<RawMessage> raws;
                
    RawMessage *meesage = raws.Add();
    meesage->put(data,len);
    CFdbParcelableBuilder builder(raws);
    if (_client)
        ((FDBusClient*)_client)->invoke(REQ_RAWDATA,builder);
}
void FDBusClientServlet::start()
{
    CBaseWorker *worker_ptr = new CBaseWorker;
    /* start worker thread */
    worker_ptr->start();
    auto client = new FDBusClient(name.c_str(),this, worker_ptr);
    client->connect(url.c_str());
    _client = client;
}
void FDBusClientServlet::wait() {

}
void FDBusClientServlet::stop() {
    CBaseWorker background_worker;
    background_worker.start(FDB_WORKER_EXE_IN_PLACE);
}
void FDBusClientServlet::onLoad(INode *node) {
    INode *child = node->childNode("connection");
    if (child) {
        name = child->getStr("name");
        url = child->getStr("url");
    }
    start();
}

void FDBusClientServlet::onMessage(ISession *session,void* data,int size)
{
    auto it = msgidHandlers.find(STREAM);
    if (it != msgidHandlers.end())
        it->second(session,data,size);
}

void FDBusClientServlet::onState(SERVLET_STATUS status,int code)
{

}

class FDBusService : public CBaseServer
{
public:
    FDBusService(const char *name,FDBusServerServlet *servlet, CBaseWorker *worker) :servlet(servlet), CBaseServer(name, worker){

    }
    ~FDBusService(){

    }

protected:
    virtual void onOnline(const CFdbOnlineInfo& info) {
        servlet->onState(OPEN,0);
    }
    virtual void onOffline(const CFdbOnlineInfo& info) {
        servlet->onState(CLOSE,0);
    }
    /* called when client calls invoke() */
    virtual void onInvoke(CBaseJob::Ptr &msg_ref)
    {
        auto msg = castToMessage<CBaseMessage *>(msg_ref);

        static int32_t elapse_time = 0;
        switch (msg->code())
        {
            case REQ_METADATA:
            {
                
            }
            break;
            case REQ_RAWDATA:
            {
                if (msg->isStatus())
                {
                    int32_t error_code;
                    std::string reason;
                    if (!msg->decodeStatus(error_code, reason))
                    {
                        FDB_LOG_E("onReply: fail to decode status!\n");
                        return;
                    }
                    LogI("onReply(): status is received: msg code: %d, error_code: %d, reason: %s\n",
                          msg->code(), error_code, reason.c_str());
                    return;
                }
                
                CFdbParcelableArray<RawMessage> raw;
                CFdbParcelableParser parser(raw);
                if (!msg->deserialize(parser))
                {
                    LogI("onReply: fail to decode from simple parser!\n");
                    return;
                }
                std::ostringstream stream;
                (void)raw.format(stream);
                
                for (auto pit = raw.pool().begin(); pit != raw.pool().end(); ++pit)
                {
             
                    std::string content;
                    content.resize(pit->size);
                    ((RawMessage*)&(*pit))->get(&content[0],pit->size);
                    FDBusSession session(msg_ref);
                    if (servlet) {
                        servlet->onMessage(&session,&content[0],content.size());
                    }
                }
                
            }
            break;
            default:
            break;
        }
    }
    /* called when client call subscribe() to register message */
    virtual void onSubscribe(CBaseJob::Ptr &msg_ref)
    {

    }

private:
    FDBusServerServlet *servlet;
};

void FDBusServerServlet::on(SERVLET_STATUS type,std::function<int(ISession *session,void*,int)> handler)
{
   msgidHandlers[type] = handler;
}
void FDBusServerServlet::request(const void *data,int len)
{

}
void FDBusServerServlet::start()
{
    CBaseWorker *worker_ptr = new CBaseWorker;
    /* start worker thread */
    worker_ptr->start();

    auto server = new FDBusService(name.c_str(), this,worker_ptr);
    server->bind(url.c_str());
    LogI("FDBusServerServlet::start %s\n",url.c_str());
}
void FDBusServerServlet::wait()
{

}
void FDBusServerServlet::stop()
{

}
void FDBusServerServlet::onLoad(INode *node)
{
    INode *child = node->childNode("connection");
    if (child) {
        name = child->getStr("name");
        url = child->getStr("url");
    }
    start();
}

void FDBusServerServlet::onState(SERVLET_STATUS status,int code)
{

}

void FDBusServerServlet::onMessage(ISession *session,void* data,int size)
{
    auto it = msgidHandlers.find(STREAM);
    if (it != msgidHandlers.end())
        it->second(session,data,size);
}

void FDBusStartup::onLoad(INode *node)
{
    FDB_CONTEXT->start();
}
void FDBusStartup::onUnload()
{

}

void FDBusEndLife::onLoad(INode *node)
{

}
void FDBusEndLife::onUnload()
{
    CBaseWorker background_worker;
    background_worker.start(FDB_WORKER_EXE_IN_PLACE);
}

REG_PLUGIN(FDBusStartup);
REG_PLUGIN(FDBusEndLife);
REG_PLUGIN(FDBusClientServlet);
REG_PLUGIN(FDBusServerServlet);